package com.lizhiyu.flink.demo6_watermark;

import com.lizhiyu.flink.model.TimeUtils;
import com.mysql.cj.util.TimeUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;

/**
 *  flink 由于提供了时间窗口，滑动，滚动(根据固定时间段来对数据分窗)
 *  1、由于数据分窗是要根据时间来对数据分窗口，这个时间可以取 1、flink处理数据的时间 (process time) 2、创建数据的时间 (event Time)
 *  2、如果数据不用必须精准则可以使用 process time，这样性能高也不用一些复杂操作
 *  3、如果使用 event time 就要考虑数据的迟到，以及数据的乱序问题
 *
 *  如果使用 event time 开窗问题
 *  1、假设有两个时间窗口，第二个窗口的数据来了，要立刻汇总第一个窗口中的数据么(触发窗口汇总的时机是什么,汇总要保证容忍情况下数据不丢失)
 *  2、假设有两个时间窗口，第一个窗口中的数据已经汇总了，这时又来一条第一个窗口的数据怎么办(窗口汇总后有哪些兜底策略)
 *
 *  解决 (触发窗口汇总的时机是什么,汇总要保证容忍时间下数据不丢失)
 *  1、引入wartwark概念(最大容忍时间)，用于计算窗口汇总时间
 *   汇总时机为  当前计算窗口最大 Event Time - 允许乱序延迟的时间 watermark >= 窗口结束时间 Window EndTime
 *
 *  解决(窗口汇总后有哪些兜底策略)
 *  1、使用allowedLateness设置二次兜底时间，进行二次兜底,当迟到数据在兜底时间内收到则重新汇总数据所在窗口然后再次输出
 *  2、超过二次兜底就要使用侧输出流了,将迟到的数据放到一个新的流中
 *
 */
public class WaterMark {

    //分类统计订单成交额
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //必须是单核处理才可以看到效果，不知道为啥
        env.setParallelism(1);
        //下载  netcat
        //(下载成功后)
        //windows nc -l -p 8888
        //输入
        //java,2022-11-11 23:12:07,10     会放到第一个窗口
        //java,2022-11-11 23:12:11,10     会放到第二个窗口    所有窗口中最大时间 23:12:11 - waterMark 3秒  >= 第一个窗口结束时间 23:12:10  才会触发第一个窗口的汇总
        //java,2022-11-11 23:12:08,10     会放到第一个窗口
        //mysql,2022-11-11 23:12:13,10    放到不同分组中,    23:12:13 - 3秒 >= 23:12:10  此时触发第一个窗口的汇总
        //java,2022-11-11 23:12:13,10     会放到第二个窗口
        //java,2022-11-11 23:12:17,10     会放到第二个窗口
        //java,2022-11-11 23:12:09,10     第一个窗口已经汇总，消息丢弃
        //java,2022-11-11 23:12:20,10     会放到第二个窗口
        //java,2022-11-11 23:12:23,10     触发第二个窗口的汇总，打印两条汇总数据，一条是java的一条是mysql的
        DataStream<String> ds = env.socketTextStream("127.0.0.1",7777);
        DataStream<Tuple3<String, String,Integer>> flatMapDS = ds.flatMap(new FlatMapFunction<String, Tuple3<String, String, Integer>>() {
            @Override
            public void flatMap(String value, Collector<Tuple3<String, String,Integer>> out) throws Exception {
                String[] arr = value.split(",");
                out.collect(Tuple3.of(arr[0], arr[1],Integer.parseInt(arr[2])));
            }
        });
        //设置warterMarker 有个WatermarkStrategy 策略
        SingleOutputStreamOperator<Tuple3<String, String,Integer>> watermakerDS = flatMapDS.assignTimestampsAndWatermarks(WatermarkStrategy
                //指定最大允许的延迟/乱序 时间
                .<Tuple3<String, String,Integer>>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(
                        (event, timestamp) -> {
                            //指定POJO的事件时间列,返回时间戳，返回这条数据对应的event time
                            return TimeUtils.strToDate(event.f1).getTime();
                        }
                ));
        SingleOutputStreamOperator<String> sumDS = watermakerDS.keyBy(new KeySelector<Tuple3<String, String,Integer>, String>() {
            @Override
            public String getKey(Tuple3<String, String,Integer> value) throws Exception {
                return value.f0;
            }
            //窗口大小为10s [0,10)  [10,20)
        }).window(TumblingEventTimeWindows.of(Time.seconds(10)))
                //二次兜底时间为1分钟
                .allowedLateness(Time.minutes(1))
                .apply(
                new WindowFunction<Tuple3<String, String,Integer>, String, String, TimeWindow>() {
            //apply 是全量汇总，相对process 没有一些上下文信息，aggregate 是增量汇总
            @Override
            public void apply(String key, TimeWindow window, Iterable<Tuple3<String, String,Integer>> input, Collector<String> out) throws Exception {
                //存放窗口的数据的事件时间
                List<String> eventTimeList = new ArrayList<>();
                int total = 0;
                for (Tuple3<String, String,Integer> order : input) {
                    eventTimeList.add(order.f1);
                    total = total+order.f2;
                }
                String outStr = String.format("分组key:%s,聚合值:%s,窗口开始结束:[%s~%s),窗口所有事件时间:%s", key,total, TimeUtils.format(window.getStart()),TimeUtils.format(window.getEnd()), eventTimeList);
                out.collect(outStr);
            }
        });
        sumDS.print();
        env.execute("watermark job");
    }
}
